package com.bw.ads;

import com.bw.util.KafkaUtil;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @BelongsProject: work_new_order
 * @BelongsPackage: com.bw.ads
 * @Author: wuxiaopeng
 * @CreateTime: 2025-07-10  15:11
 * @Description: TODO
 * @Version: 1.0
 */
public class 店铺动态人气统计 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        DataStreamSource<String> df = KafkaUtil.getKafkaSource(env, "big_table", "asdf");
//        df.print();
        DataStreamSource<String> cart = KafkaUtil.getKafkaSource(env, "new_work_order_dwd_cart", "asdf");
//        cart.print();

//      TODO 加购表的创建
        tenv.executeSql("CREATE TABLE cart (\n" +
                "    id bigint ,\n" +
                "    user_id varchar,\n" +
                "    sku_id bigint,\n" +
                "    cart_price decimal(10,2),\n" +
                "    sku_num int,\n" +
                "    create_time bigint,\n" +
                "    operate_time bigint,\n" +
                "    is_ordered bigint ,\n" +
                "    order_time bigint,\n" +
                "     time_ltz AS TO_TIMESTAMP_LTZ(create_time, 3),\n" +
                "     WATERMARK FOR time_ltz AS time_ltz - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                "'connector' = 'kafka',\n" +
                "'topic' = 'new_work_order_dwd_cart',\n" +
                "'properties.bootstrap.servers' = 'hadoop101:9092',\n" +
                "'properties.group.id' = 'testGroup',\n" +
                "'scan.startup.mode' = 'earliest-offset',\n" +
                "'format' = 'json'\n" +
                ")");
        Table cartTable = tenv.sqlQuery("select * from cart");
//        tenv.toChangelogStream(cartTable).print();

//      TODO 加购人数、支付买家数和支付金额
        Table add = tenv.sqlQuery("select \n" +
                "             user_id,\n" +
                "        count(*) as add_num,\n" +
                "        count(DISTINCT case when is_ordered=1 then user_id end) as pay_add_num, \n" +
                "        sum(if(is_ordered=1,cart_price*sku_num,0)) as pay_price \n" +
                "        from cart group by user_id");

        DataStream<Row> tomysql_add = tenv.toChangelogStream(add);

//        tomysql_add.print();

        Table user_num = tenv.sqlQuery("select \n" +
                " count(distinct user_id) as user_num from cart");

        DataStream<Row> user = tenv.toChangelogStream(user_num);
//        user.print();


//        tomysql_add.addSink(
//                JdbcSink.sink(
//                        "insert into order_add (user_id,add_num, pay_add_num, pay_price) values (?, ?, ?, ?)",
//                        (statement, book) -> {
//                            statement.setLong(1, new Long(book.getField("user_id")+""));
//                            statement.setLong(2, new Long(book.getField("add_num")+""));
//                            statement.setLong(3, new Long(book.getField("pay_add_num")+""));
//                            statement.setDouble(4, new Double(book.getField("pay_price")+""));
//                        },
//                        JdbcExecutionOptions.builder()
//                                .withBatchSize(1)
//                                .withBatchIntervalMs(200)
//                                .withMaxRetries(5)
//                                .build(),
//                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
//                                .withUrl("jdbc:mysql://hadoop101:3306/work_order_FineBi?useSSL=false&serverTimezone=UTC")
//                                .withDriverName("com.mysql.cj.jdbc.Driver") // MySQL 8.x驱动
//                                .withUsername("root")
//                                .withPassword("000000")
//                                .build()
//                ));

//                user.addSink(
//                JdbcSink.sink(
//                        "insert into user_num (user_num) values (?)",
//                        (statement, book) -> {
//                            statement.setLong(1, new Long(book.getField("user_num")+""));
//                        },
//                        JdbcExecutionOptions.builder()
//                                .withBatchSize(1)
//                                .withBatchIntervalMs(200)
//                                .withMaxRetries(5)
//                                .build(),
//                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
//                                .withUrl("jdbc:mysql://hadoop101:3306/work_order_FineBi?useSSL=false&serverTimezone=UTC")
//                                .withDriverName("com.mysql.cj.jdbc.Driver") // MySQL 8.x驱动
//                                .withUsername("root")
//                                .withPassword("000000")
//                                .build()
//                ));


//        销售额TOP10门店
        tenv.executeSql("CREATE TABLE big_data (\n" +
                "    id bigint,\n" +
                "    consignee STRING,\n" +
                "    consigneeTel STRING,\n" +
                "    totalAmount DECIMAL(10, 2),\n" +
                "    orderStatus STRING,\n" +
                "    userId BIGINT,\n" +
                "    paymentWay STRING,\n" +
                "    shop_id BIGINT,\n" +
                "    deliverAddress STRING,\n" +
                "    orderComment STRING,\n" +
                "    outTradeNo STRING,\n" +
                "    tradeBody STRING,\n" +
                "    createTime bigint,\n" +
                "    processStatus STRING,\n" +
                "    trackingNo STRING,\n" +
                "    parentOrderId BIGINT,\n" +
                "    provinceId INT,\n" +
                "    activityReduceAmount DECIMAL(10, 2),\n" +
                "    couponReduceAmount DECIMAL(10, 2),\n" +
                "    originalTotalAmount DECIMAL(10, 2),\n" +
                "    freightFee DECIMAL(10, 2),\n" +
                "    freightFeeReduce DECIMAL(10, 2),\n" +
                "    orderId BIGINT,\n" +
                "    skuId BIGINT,\n" +
                "    orderPrice DECIMAL(10, 2),\n" +
                "    skuNum BIGINT,\n" +
                "    splitTotalAmount DECIMAL(10, 2),\n" +
                "    splitActivityAmount DECIMAL(10, 2),\n" +
                "    splitCouponAmount DECIMAL(10, 2),\n" +
                "    spuId BIGINT,\n" +
                "    price INT,\n" +
                "    skuName STRING,\n" +
                "    taid BIGINT,\n" +
                "    category3Id BIGINT,\n" +
                "    isSale INT,\n" +
                "    spuName STRING,\n" +
                "    userName STRING,\n" +
                "    phoneNum STRING,\n" +
                "    birthday STRING,\n" +
                "    gender STRING,\n" +
                "    c3name STRING,\n" +
                "    category2Id BIGINT,\n" +
                "    c2name STRING,\n" +
                "    category1Id BIGINT,\n" +
                "    c1name STRING,\n" +
                "    provinceName STRING,\n" +
                "    regionId BIGINT,\n" +
                "    areaCode STRING,\n" +
                "    iso3661 STRING,\n" +
                "    iso31662 STRING,\n" +
                "    regionName STRING,\n" +
                "    shopName STRING,\n" +
                " time_ltz AS TO_TIMESTAMP_LTZ(createTime, 3),\n" +
                " WATERMARK FOR time_ltz AS time_ltz - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'big_table',\n" +
                "  'properties.bootstrap.servers' = 'hadoop101:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'earliest-offset',\n" +
                "  'format' = 'json'\n" +
                ")");

        Table big_data = tenv.sqlQuery("select * from big_data");
//        tenv.toChangelogStream(big_data).print();


        //        销售额TOP10门店
        Table top = tenv.sqlQuery("SELECT \n" +
                "shop_id,shopName,\n" +
                "SUM(totalAmount) AS total_sales\n" +
                "FROM\n" +
                "big_data\n" +
                "GROUP BY\n" +
                "shop_id,shopName\n" +
                "ORDER BY\n" +
                "total_sales DESC \n" +
                "LIMIT 5");
        DataStream<Row> top5 = tenv.toChangelogStream(top);
//        top5.print();


//        top5.addSink(
//                JdbcSink.sink(
//                        "insert into shop_top5 (shop_id,shopName,total_sales) values (?,?,?)",
//                        (statement, book) -> {
//                            statement.setLong(1, new Long(book.getField("shop_id")+""));
//                            statement.setString(2, book.getField("shopName")+"");
//                            statement.setDouble(3, new Double(book.getField("total_sales")+""));
//                        },
//                        JdbcExecutionOptions.builder()
//                                .withBatchSize(1)
//                                .withBatchIntervalMs(200)
//                                .withMaxRetries(5)
//                                .build(),
//                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
//                                .withUrl("jdbc:mysql://hadoop101:3306/work_order_FineBi?useSSL=false&serverTimezone=UTC")
//                                .withDriverName("com.mysql.cj.jdbc.Driver") // MySQL 8.x驱动
//                                .withUsername("root")
//                                .withPassword("000000")
//                                .build()
//                ));

        Table month = tenv.sqlQuery("select FROM_UNIXTIME(createTime / 1000, 'yyyy-MM-dd') AS year_months,sum(totalAmount) sum_amount from big_data group by FROM_UNIXTIME(createTime / 1000, 'yyyy-MM-dd')");
        DataStream<Row> month_amount1 = tenv.toChangelogStream(month);
//        month_amount1.print();

//        month_amount1.addSink(
//                JdbcSink.sink(
//                        "insert into month_amount (year_months,sum_amount) values (?, ?)",
//                        (statement, book) -> {
//                            statement.setString(1, book.getField("year_months")+"");
//                            statement.setDouble(2, new Double(book.getField("sum_amount")+""));
//                        },
//                        JdbcExecutionOptions.builder()
//                                .withBatchSize(1)
//                                .withBatchIntervalMs(200)
//                                .withMaxRetries(5)
//                                .build(),
//                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
//                                .withUrl("jdbc:mysql://hadoop101:3306/work_order_FineBi?useSSL=false&serverTimezone=UTC")
//                                .withDriverName("com.mysql.cj.jdbc.Driver") // MySQL 8.x驱动
//                                .withUsername("root")
//                                .withPassword("000000")
//                                .build()
//                ));

        Table day_shop_amount = tenv.sqlQuery("select " +
                "shop_id,shopName,sum(totalAmount) as total_sales,FROM_UNIXTIME(createTime / 1000, 'yyyy-MM-dd') AS year_months " +
                "from big_data where shop_id=1111 or shop_id=1113" +
                " group by shop_id,shopName,FROM_UNIXTIME(createTime / 1000, 'yyyy-MM-dd')");
        DataStream<Row> day_shop_amount_tomysql = tenv.toChangelogStream(day_shop_amount);
//        day_shop_amount_tomysql.print();

//        day_shop_amount_tomysql.addSink(
//                JdbcSink.sink(
//                        "insert into day_shop_amount (shop_id,shopName,total_sales,year_months) values (?, ?,?, ?)",
//                        (statement, book) -> {
//                            statement.setLong(1, new Long(book.getField("shop_id")+""));
//                            statement.setString(2,book.getField("shopName")+"");
//                            statement.setDouble(3, new Double(book.getField("total_sales")+""));
//                            statement.setString(4, book.getField("year_months")+"");
//                        },
//                        JdbcExecutionOptions.builder()
//                                .withBatchSize(1)
//                                .withBatchIntervalMs(200)
//                                .withMaxRetries(5)
//                                .build(),
//                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
//                                .withUrl("jdbc:mysql://hadoop101:3306/work_order_FineBi?useSSL=false&serverTimezone=UTC")
//                                .withDriverName("com.mysql.cj.jdbc.Driver") // MySQL 8.x驱动
//                                .withUsername("root")
//                                .withPassword("000000")
//                                .build()
//                ));


        Table province_amount = tenv.sqlQuery("select " +
                "provinceName,sum(totalAmount) as total_sales " +
                "from big_data " +
                "group by provinceName");
        DataStream<Row> province_amount_tomysql = tenv.toChangelogStream(province_amount);
//        province_amount_tomysql.print();


//        province_amount_tomysql.addSink(
//                JdbcSink.sink(
//                        "insert into province_amount (provinceName,total_sales) values (?, ?)",
//                        (statement, book) -> {
//                            statement.setString(1, book.getField("provinceName")+"");
//                            statement.setDouble(2, new Double(book.getField("total_sales")+""));
//                        },
//                        JdbcExecutionOptions.builder()
//                                .withBatchSize(1)
//                                .withBatchIntervalMs(200)
//                                .withMaxRetries(5)
//                                .build(),
//                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
//                                .withUrl("jdbc:mysql://hadoop101:3306/work_order_FineBi?useSSL=false&serverTimezone=UTC")
//                                .withDriverName("com.mysql.cj.jdbc.Driver") // MySQL 8.x驱动
//                                .withUsername("root")
//                                .withPassword("000000")
//                                .build()
//                ));

        Table shop_sales = tenv.sqlQuery("SELECT " +
                "shop_id, " +
                "shopName, " +
                "SUM(CASE WHEN FROM_UNIXTIME(createTime / 1000, 'yyyy-MM-dd') >= CURRENT_TIMESTAMP - INTERVAL '1' DAY THEN totalAmount ELSE 0 END) AS sales_1_day, " +
                "SUM(CASE WHEN FROM_UNIXTIME(createTime / 1000, 'yyyy-MM-dd') >= CURRENT_TIMESTAMP - INTERVAL '7' DAY THEN totalAmount ELSE 0 END) AS sales_7_days, " +
                "SUM(CASE WHEN FROM_UNIXTIME(createTime / 1000, 'yyyy-MM-dd') >= CURRENT_TIMESTAMP - INTERVAL '30' DAY THEN totalAmount ELSE 0 END) AS sales_30_days " +
                "FROM big_data " +
                "GROUP BY shop_id, shopName");

        DataStream<Row> shop_sales_stream = tenv.toChangelogStream(shop_sales);
        shop_sales_stream.print();

//        shop_sales_stream.addSink(
//                JdbcSink.sink(
//                        "insert into date_shop_amount (shop_id,shopName, sales_1_day, sales_7_days, sales_30_days) values (?, ?, ?, ?, ?)",
//                        (statement, book) -> {
//                            statement.setLong(1, new Long(book.getField("shop_id") + ""));
//                            statement.setString(2, book.getField("shopName") + "");
//                            statement.setDouble(3, new Double(book.getField("sales_1_day") + ""));
//                            statement.setDouble(4, new Double(book.getField("sales_7_days") + ""));
//                            statement.setDouble(5, new Double(book.getField("sales_30_days") + ""));
//                        },
//                        JdbcExecutionOptions.builder()
//                                .withBatchSize(1)
//                                .withBatchIntervalMs(200)
//                                .withMaxRetries(5)
//                                .build(),
//                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
//                                .withUrl("jdbc:mysql://hadoop101:3306/work_order_FineBi?useSSL=false&serverTimezone=UTC")
//                                .withDriverName("com.mysql.cj.jdbc.Driver") // MySQL 8.x驱动
//                                .withUsername("root")
//                                .withPassword("000000")
//                                .build()
//                ));














        env.execute("work_order");
    }
}
